package com.tg.base.concurrent.future;

import java.util.Objects;
import java.util.concurrent.*;

/**
 * Created by MagicBook on 2020/2/2.
 */
public class CompletableFutureApiTest {
    public static void main(String[] args) {
        try {
//            thenCombine();
            thenCombineAndThenNotify();
//            asyncAgain();
//            anyOfallOf();

//            asyncAgain();
//            runAfterBoth();
//            thenAcceptBoth();
//            thenCombine();
        } catch (InterruptedException | ExecutionException |TimeoutException ex) {
            ex.printStackTrace();
        }


    }

    /**
     * 异步再异步测试
     * 结果
     * run 2
     * done 2
     * run 3
     * done 3
     */
    private static void asyncAgain()
            throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("run 1");
            threadSleep(1000 * 120);
            return 1;
        });

        future.thenApplyAsync((a) -> {
            System.out.println("run 2:" + a);
            return 2;
        });

        future.thenApplyAsync((a) -> {
            System.out.println("run 3:" + a);
            return 3;
        });

        threadSleep(500);
    }

    /**
     * allOf()：所有future完成执行，不能传递future结果到后续方法，配合join()使用可以阻塞等待
     * anyOf()：任意一个future完成，传递第一个完成future结果到后续方法，配合get()阻塞等待第一个结果完成
     */
    private static void anyOfallOf()
            throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
            threadSleep(100);
            return 1;
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            threadSleep(200);
            return 2;
        });

        CompletableFuture.anyOf(future1, future2).thenAccept((r) -> {
            System.out.println("result:" + Objects.equals(r, 1));
        });
        System.out.println("done");

        CompletableFuture.anyOf(future1, future2).thenAccept((r) -> {
            System.out.println("with get() result:" + Objects.equals(r, 1));
        }).get();

        System.out.println("with get() done");
    }

    /**
     * 运行于两个future之后
     */
    private static void runAfterBoth()
            throws InterruptedException, ExecutionException {
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            System.out.println("future1");
        });
        CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
            System.out.println("future2");
        });

        future1.runAfterBoth(future2, () -> {
            System.out.println("runAfterBoth");
        });
    }

    /**
     * 同时接受两个future的结果
     *
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void thenAcceptBoth()
            throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
            return 1;
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            return 2;
        });

        future1.thenAcceptBoth(future2, (r1, r2) -> {
            System.out.println("r1:" + r1);
            System.out.println("r2:" + r2);
        });

        threadSleep(100);
    }

    /**
     * 同时接受两个future结果，并返回新结果
     */
    private static void thenCombine()
            throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
            threadSleep(2000);
            return 1;
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            threadSleep(1000);
            return 2;
        });

        CompletableFuture<Integer> resultFuture = future1.thenCombine(future2, (a, b) -> {
            Integer c = a + b;
            System.out.println(c);
            return c;
        });

        //立即返回结果，实现future全部能力
        Integer target = resultFuture.getNow(-1);
        System.out.println(target);

        //
        Integer number = resultFuture.getNumberOfDependents();
        System.out.println(number);

        threadSleep(5000);
    }

    /**
     * 同时接受两个future结果，并返回新结果，并异步通知
     */
    private static void thenCombineAndThenNotify()
            throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
            threadSleep(2000);
            return 1;
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            threadSleep(1000);
            return 2;
        });

        CompletableFuture<Integer> resultFuture = future1.thenCombine(future2, (a, b) -> {
            Integer c = a + b;
            System.out.println(Thread.currentThread().getName() +":"+c);
            return c;
        });

        Executor executor = Executors.newFixedThreadPool(1);
        resultFuture.thenRunAsync(() -> {
            System.out.println(Thread.currentThread().getName() + ":notify each other");
        }, executor);

        Integer target = future1.get(4, TimeUnit.SECONDS);
        System.out.println(Thread.currentThread().getName() +":"+target);


        threadSleep(5000);
    }

    private static void threadSleep(long ms) {
        try {
            TimeUnit.MILLISECONDS.sleep(ms);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
    }
}
